package rds

import (
	"context"
	"sync"
	"time"

	"github.com/cloudwego/hertz/pkg/common/hlog"
	"github.com/redis/go-redis/v9"
)

// Message 消息结构体
type Message struct {
	Channel      string
	Pattern      string
	Payload      string
	PayloadSlice []string
	ReceivedAt   time.Time
}

// MessageHandler 消息处理器接口
type MessageHandler func(ctx context.Context, msg *Message) error

// Subscriber 订阅者接口
type Subscriber interface {
	// Subscribe 订阅指定频道
	Subscribe(ctx context.Context, channels ...string) error
	// PSubscribe 模式订阅
	PSubscribe(ctx context.Context, patterns ...string) error
	// Unsubscribe 取消订阅
	Unsubscribe(ctx context.Context, channels ...string) error
	// PUnsubscribe 取消模式订阅
	PUnsubscribe(ctx context.Context, patterns ...string) error
	// Close 关闭订阅者
	Close() error
	// IsClosed 检查是否已关闭
	IsClosed() bool
}

// Publisher 发布者接口
type Publisher interface {
	// Publish 发布消息到指定频道
	Publish(ctx context.Context, channel string, message interface{}) error
	// PublishWithTTL 发布消息并设置过期时间
	PublishWithTTL(ctx context.Context, channel string, message interface{}, ttl time.Duration) error
}

// PubSubManager Redis发布订阅管理器
type PubSubManager struct {
	client     *Client
	subscriber *redis.PubSub
	handlers   map[string]MessageHandler
	patterns   map[string]MessageHandler
	mu         sync.RWMutex
	ctx        context.Context
	cancel     context.CancelFunc
	wg         sync.WaitGroup
	closed     bool
	listening  bool // 添加监听状态标志
}

// NewPubSubManager 创建新的发布订阅管理器
func NewPubSubManager(client *Client) *PubSubManager {
	ctx, cancel := context.WithCancel(context.Background())
	return &PubSubManager{
		client:   client,
		handlers: make(map[string]MessageHandler),
		patterns: make(map[string]MessageHandler),
		ctx:      ctx,
		cancel:   cancel,
	}
}

// Subscribe 订阅指定频道
func (psm *PubSubManager) Subscribe(ctx context.Context, channels ...string) error {
	psm.mu.Lock()
	defer psm.mu.Unlock()

	if psm.subscriber == nil {
		psm.subscriber = psm.client.Subscribe(ctx, channels...)
	} else {
		err := psm.subscriber.Subscribe(ctx, channels...)
		if err != nil {
			return err
		}
	}

	// 只在第一次订阅且未开始监听时启动消息监听
	if !psm.closed && !psm.listening && psm.subscriber != nil {
		psm.listening = true
		psm.wg.Add(1)
		go psm.listenMessages()
	}

	return nil
}

// PSubscribe 模式订阅
func (psm *PubSubManager) PSubscribe(ctx context.Context, patterns ...string) error {
	psm.mu.Lock()
	defer psm.mu.Unlock()

	if psm.subscriber == nil {
		psm.subscriber = psm.client.PSubscribe(ctx, patterns...)
	} else {
		err := psm.subscriber.PSubscribe(ctx, patterns...)
		if err != nil {
			return err
		}
	}

	// 只在第一次订阅且未开始监听时启动消息监听
	if !psm.closed && !psm.listening && psm.subscriber != nil {
		psm.listening = true
		psm.wg.Add(1)
		go psm.listenMessages()
	}

	return nil
}

// Unsubscribe 取消订阅
func (psm *PubSubManager) Unsubscribe(ctx context.Context, channels ...string) error {
	psm.mu.Lock()
	defer psm.mu.Unlock()

	if psm.subscriber != nil {
		return psm.subscriber.Unsubscribe(ctx, channels...)
	}
	return nil
}

// PUnsubscribe 取消模式订阅
func (psm *PubSubManager) PUnsubscribe(ctx context.Context, patterns ...string) error {
	psm.mu.Lock()
	defer psm.mu.Unlock()

	if psm.subscriber != nil {
		return psm.subscriber.PUnsubscribe(ctx, patterns...)
	}
	return nil
}

// Close 关闭订阅者
func (psm *PubSubManager) Close() error {
	psm.mu.Lock()
	defer psm.mu.Unlock()

	if psm.closed {
		return nil
	}

	psm.closed = true
	psm.listening = false // 重置监听状态

	// 先取消上下文，通知监听协程停止
	psm.cancel()

	// 等待监听协程结束
	psm.wg.Wait()

	// 关闭Redis订阅连接
	if psm.subscriber != nil {
		err := psm.subscriber.Close()
		psm.subscriber = nil
		if err != nil {
			return err
		}
	}

	return nil
}

// IsClosed 检查是否已关闭
func (psm *PubSubManager) IsClosed() bool {
	psm.mu.RLock()
	defer psm.mu.RUnlock()
	return psm.closed
}

// AddHandler 添加频道消息处理器
func (psm *PubSubManager) AddHandler(channel string, handler MessageHandler) {
	psm.mu.Lock()
	defer psm.mu.Unlock()
	psm.handlers[channel] = handler
}

// AddPatternHandler 添加模式消息处理器
func (psm *PubSubManager) AddPatternHandler(pattern string, handler MessageHandler) {
	psm.mu.Lock()
	defer psm.mu.Unlock()
	psm.patterns[pattern] = handler
}

// RemoveHandler 移除频道消息处理器
func (psm *PubSubManager) RemoveHandler(channel string) {
	psm.mu.Lock()
	defer psm.mu.Unlock()
	delete(psm.handlers, channel)
}

// RemovePatternHandler 移除模式消息处理器
func (psm *PubSubManager) RemovePatternHandler(pattern string) {
	psm.mu.Lock()
	defer psm.mu.Unlock()
	delete(psm.patterns, pattern)
}

// listenMessages 监听消息
func (psm *PubSubManager) listenMessages() {
	defer func() {
		psm.wg.Done()
		// 重置监听状态
		psm.mu.Lock()
		psm.listening = false
		psm.mu.Unlock()
	}()

	for {
		select {
		case <-psm.ctx.Done():
			return
		default:
			if psm.subscriber == nil {
				return
			}

			msg, err := psm.subscriber.ReceiveMessage(psm.ctx)
			if err != nil {
				if err == context.Canceled {
					return
				}
				// 检查是否是连接关闭错误
				if err.Error() == "use of closed network connection" ||
					err.Error() == "read tcp: use of closed network connection" {
					hlog.Debugf("Redis连接已关闭，停止监听: %v", err)
					return
				}
				hlog.Errorf("receive message error: %v", err)
				// 添加短暂延迟避免无限循环
				time.Sleep(100 * time.Millisecond)
				continue
			}

			message := &Message{
				Channel:    msg.Channel,
				Pattern:    msg.Pattern,
				Payload:    msg.Payload,
				ReceivedAt: time.Now(),
			}

			psm.handleMessage(message)
		}
	}
}

// handleMessage 处理消息
func (psm *PubSubManager) handleMessage(msg *Message) {
	psm.mu.RLock()
	defer psm.mu.RUnlock()

	// 优先处理模式匹配的消息
	if msg.Pattern != "" {
		if handler, exists := psm.patterns[msg.Pattern]; exists {
			go func() {
				if err := handler(psm.ctx, msg); err != nil {
					hlog.Errorf("pattern handler error for %s: %v", msg.Pattern, err)
				}
			}()
			return
		}
	}

	// 处理频道消息
	if handler, exists := psm.handlers[msg.Channel]; exists {
		go func() {
			if err := handler(psm.ctx, msg); err != nil {
				hlog.Errorf("channel handler error for %s: %v", msg.Channel, err)
			}
		}()
	}
}

// Publish 发布消息到指定频道
func (psm *PubSubManager) Publish(ctx context.Context, channel string, message interface{}) error {
	return psm.client.Publish(ctx, channel, message).Err()
}

// PublishWithTTL 发布消息并设置过期时间
func (psm *PubSubManager) PublishWithTTL(ctx context.Context, channel string, message interface{}, ttl time.Duration) error {
	// 使用管道批量执行
	pipe := psm.client.Pipeline()
	pipe.Publish(ctx, channel, message)
	pipe.Expire(ctx, channel, ttl)

	_, err := pipe.Exec(ctx)
	return err
}

// GetSubscriber 获取订阅者接口
func (psm *PubSubManager) GetSubscriber() Subscriber {
	return psm
}

// GetPublisher 获取发布者接口
func (psm *PubSubManager) GetPublisher() Publisher {
	return psm
}
